package com.shujia.card

import java.text.SimpleDateFormat
import java.util.Date

import com.shujia.util.{CarUtil, SparkStreamTool}
import org.apache.spark.streaming.dstream.DStream
import redis.clients.jedis.Jedis

object RealTimeRoadDayFlow extends SparkStreamTool {
  override def run(args: Array[String]): Unit = {

    ///读取数据

    val carsDS: DStream[CarUtil.Car] = CarUtil.loadKafkaCar(ssc, "RealTimeRoadDayFlow")

    /**
      * 实时统计每个道路当天总流量
      *
      */

    val kvDS: DStream[(String, Int)] = carsDS.map(car => {
      val road_id: Long = car.road_id
      val time: Long = car.time
      val date = new Date(time * 1000)
      val format = new SimpleDateFormat("yyyyMMdd")
      val day: String = format.format(date)

      (road_id + ":" + day, 1)
    })

    val countDS: DStream[(String, Int)] = kvDS.updateStateByKey((seq: Seq[Int], opt: Option[Int]) => Option(seq.sum + opt.getOrElse(0)))

    //将统计结果保存到redis中
    countDS.foreachRDD(rdd => {
      rdd.foreachPartition(iter => {

        val jedis = new Jedis("master", 6379)

        iter.foreach {
          case (roadAndDay: String, num: Int) =>
            jedis.set("RealTimeRoadDayFlow:" + roadAndDay, num.toString)
        }

        jedis.close()
      })
    })

  }
}
